package cn.itcast.flink.explain;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class ExplainSqlTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        DataStream<Tuple2<Integer, String>> stream1 = env.fromElements(new Tuple2<>(1, "hello"));
        DataStream<Tuple2<Integer, String>> stream2 = env.fromElements(new Tuple2<>(1, "hello"));

        // explain Table API
        Table table1 = tEnv.fromDataStream(stream1, $("count"), $("word"));
        Table table2 = tEnv.fromDataStream(stream2, $("count"), $("word"));
        // FlinkTable 执行计划的查询
        // ① table 表对象 . explain
        Table table = table1
                .where($("word").like("F%"))
                .unionAll(table2);

        System.out.println(table.explain());
        // ② Flinksql tEnv.explainSql(执行SQL)
        tEnv.createTemporaryView("t_t1",table1);
        tEnv.createTemporaryView("t_t2",table2);
        tEnv.explainSql("select `count`,word from t_t1 " +
                "where word like 'F%' " +
                "UNION ALL select `count`,word" +
                " from t_t2");

        env.execute();
    }
}